#include "pth.h"

// 元宇宙上报线程：周期性从共享内存读取数据并通过MQTT上报
void *pthread_yuanyuzhou(void *arg)
{
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    int rc;

    // 创建MQTT客户端并连接
    MQTTClient_create(&client, ADDRESS, CLIENTID_YUANYU,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to pthreadpublic connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    printf("元宇宙上报线程ready\n");

    char payload[64];
    int *shm_stm_addr_int = (int *)shm_stm_addr;
    Shm_Stm *shm_stm_addr_struct = (Shm_Stm *)(shm_stm_addr_int + 1);
    
    while (1)
    {
        memset(payload, 0, 64);
        cJSON *root = cJSON_CreateObject(); // 每次循环重新创建root，避免累积
        
        for (int i = 0; i < *shm_stm_addr_int; i++)
        {
            switch (shm_stm_addr_struct->key)
            {
            case 1:
                cJSON_AddItemToObject(root, "tem", cJSON_CreateNumber(shm_stm_addr_struct->val.c));
                break;
            case 2:
                cJSON_AddItemToObject(root, "hum", cJSON_CreateNumber(shm_stm_addr_struct->val.c));
                break;
            case 3:
                cJSON_AddItemToObject(root, "infrared", cJSON_CreateNumber(
                    shm_stm_addr_struct->val.a == 1 ? 31 : 30));
                break;
            case 4:
                cJSON_AddItemToObject(root, "flame", cJSON_CreateBool(shm_stm_addr_struct->val.a));
                break;
            case 5:
                cJSON_AddItemToObject(root, "fan", cJSON_CreateBool(shm_stm_addr_struct->val.a));
                break;
            case 6:
                cJSON_AddItemToObject(root, "sunshade", cJSON_CreateString(
                    shm_stm_addr_struct->val.a == 1 ? "forward" : "reverse"));
                break;
            case 7:
                cJSON_AddItemToObject(root, "lamp", cJSON_CreateBool(shm_stm_addr_struct->val.a));
                break;
            case 8:
                cJSON_AddItemToObject(root, "alarm", cJSON_CreateBool(shm_stm_addr_struct->val.a));
                break;
            default:
                break;
            }

            strcpy(payload, cJSON_Print(root));
            pubmsg.payload = payload;
            pubmsg.payloadlen = (int)strlen(payload);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;
            MQTTClient_publishMessage(client, TOPICPUB, &pubmsg, &token);
            printf("Waiting for up to %d seconds for publication of %s\n"
                   "on topic %s for client with ClientID: %s\n",
                   (int)(TIMEOUT / 1000), payload, TOPICPUB, CLIENTID);

            rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
            printf("Message with delivery token %d delivered\n", token);

            shm_stm_addr_struct++;
        }
        // 重置共享内存指针
        shm_stm_addr_struct = (Shm_Stm *)(shm_stm_addr_int + 1);
        cJSON_Delete(root); // 释放当前循环创建的cJSON对象
        sleep(1);
    }
}

// 上报线程（支持变化上报和周期上报）
void *pthread_upload(void *arg)
{
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    int rc;

    MQTTClient_create(&client, ADDRESS, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to pthreadpublic connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    printf("上报线程ready\n");
    Shm_Report *shm_report_addr_struct = (Shm_Report *)shm_report_addr;

    if (shm_report_addr_struct->mode == 1) // 变化上报
    {
        printf("变化上报\n");
        // 初始化链表
        struct list_head stm_head;
        INIT_LIST_HEAD(&stm_head);
        struct list_head modbus_head;
        INIT_LIST_HEAD(&modbus_head);

        char payload[1024];
        cJSON *root = cJSON_CreateObject(); // 创建初始JSON对象

        // 处理STM数据
        {
            cJSON *item = cJSON_CreateObject();
            cJSON *arr = cJSON_CreateArray();
            cJSON_AddItemToObject(root, "m0", item);
            cJSON_AddItemToObject(item, "data", arr);

            int *shm_stm_addr_int = (int *)shm_stm_addr;
            Shm_Stm *shm_stm_addr_struct = (Shm_Stm *)(shm_stm_addr_int + 1);
            printf("%d\n", *shm_stm_addr_int);

            for (int i = 0; i < *shm_stm_addr_int; i++)
            {
                cJSON *data = cJSON_CreateObject();
                cJSON_AddItemToObject(data, "key", cJSON_CreateNumber(shm_stm_addr_struct->key));
                // 设置名称
                switch (shm_stm_addr_struct->key)
                {
                case 1: cJSON_AddItemToObject(data, "name", cJSON_CreateString("temperature")); break;
                case 2: cJSON_AddItemToObject(data, "name", cJSON_CreateString("humidity")); break;
                case 3: cJSON_AddItemToObject(data, "name", cJSON_CreateString("red_light")); break;
                case 4: cJSON_AddItemToObject(data, "name", cJSON_CreateString("fire")); break;
                case 5: cJSON_AddItemToObject(data, "name", cJSON_CreateString("fan")); break;
                case 6: cJSON_AddItemToObject(data, "name", cJSON_CreateString("window")); break;
                case 7: cJSON_AddItemToObject(data, "name", cJSON_CreateString("light")); break;
                case 8: cJSON_AddItemToObject(data, "name", cJSON_CreateString("buzzer")); break;
                default: break;
                }
                cJSON_AddItemToObject(data, "type", cJSON_CreateNumber(shm_stm_addr_struct->type));
                // 设置值（根据类型）
                if (shm_stm_addr_struct->type == 1)
                {
                    cJSON_AddItemToObject(data, "val", cJSON_CreateNumber(shm_stm_addr_struct->val.a));
                    List_Data listdata = {shm_stm_addr_struct->key, shm_stm_addr_struct->type, shm_stm_addr_struct->val.a};
                    list_add_tail(&listdata.list, &stm_head);
                }
                else if (shm_stm_addr_struct->type == 2)
                {
                    cJSON_AddItemToObject(data, "val", cJSON_CreateNumber(shm_stm_addr_struct->val.b));
                    List_Data listdata = {shm_stm_addr_struct->key, shm_stm_addr_struct->type, shm_stm_addr_struct->val.b};
                    list_add_tail(&listdata.list, &stm_head);
                }
                else if (shm_stm_addr_struct->type == 3)
                {
                    cJSON_AddItemToObject(data, "val", cJSON_CreateNumber(shm_stm_addr_struct->val.c));
                    List_Data listdata = {shm_stm_addr_struct->key, shm_stm_addr_struct->type, shm_stm_addr_struct->val.c};
                    list_add_tail(&listdata.list, &stm_head);
                }
                cJSON_AddItemToArray(arr, data);
                shm_stm_addr_struct++;
            }
        }

        // 处理Modbus数据
        {
            cJSON *item = cJSON_CreateObject();
            cJSON *arr = cJSON_CreateArray();
            cJSON_AddItemToObject(root, "modbus", item);
            cJSON_AddItemToObject(item, "data", arr);

            int *shm_modbus_addr_int = (int *)shm_modbus_addr;
            Shm_Modbus *shm_modbus_addr_struct = (Shm_Modbus *)(shm_modbus_addr_int + 1);
            printf("%d\n", *shm_modbus_addr_int);

            for (int i = 0; i < *shm_modbus_addr_int; i++)
            {
                cJSON *data = cJSON_CreateObject();
                cJSON_AddItemToObject(data, "key", cJSON_CreateNumber(shm_modbus_addr_struct->key));
                // 设置名称（修复原代码switch(1)的逻辑错误）
                switch (shm_modbus_addr_struct->key)
                {
                case 101: cJSON_AddItemToObject(data, "name", cJSON_CreateString("people")); break;
                case 102: cJSON_AddItemToObject(data, "name", cJSON_CreateString("door")); break;
                case 103: cJSON_AddItemToObject(data, "name", cJSON_CreateString("window")); break;
                case 104: cJSON_AddItemToObject(data, "name", cJSON_CreateString("air_switch")); break;
                case 105: cJSON_AddItemToObject(data, "name", cJSON_CreateString("air_temp")); break;
                case 106: cJSON_AddItemToObject(data, "name", cJSON_CreateString("fan")); break;
                default: break;
                }
                cJSON_AddItemToObject(data, "type", cJSON_CreateNumber(shm_modbus_addr_struct->type));
                // 设置值（根据类型）
                if (shm_modbus_addr_struct->type == 1)
                {
                    cJSON_AddItemToObject(data, "val", cJSON_CreateNumber(shm_modbus_addr_struct->val.a));
                    List_Data listdata = {shm_modbus_addr_struct->key, shm_modbus_addr_struct->type, shm_modbus_addr_struct->val.a};
                    list_add_tail(&listdata.list, &modbus_head);
                }
                else if (shm_modbus_addr_struct->type == 2)
                {
                    cJSON_AddItemToObject(data, "val", cJSON_CreateNumber(shm_modbus_addr_struct->val.b));
                    List_Data listdata = {shm_modbus_addr_struct->key, shm_modbus_addr_struct->type, shm_modbus_addr_struct->val.b};
                    list_add_tail(&listdata.list, &modbus_head);
                }
                else if (shm_modbus_addr_struct->type == 3)
                {
                    cJSON_AddItemToObject(data, "val", cJSON_CreateNumber(shm_modbus_addr_struct->val.c));
                    List_Data listdata = {shm_modbus_addr_struct->key, shm_modbus_addr_struct->type, shm_modbus_addr_struct->val.c};
                    list_add_tail(&listdata.list, &modbus_head);
                }
                cJSON_AddItemToArray(arr, data);
                shm_modbus_addr_struct++;
            }
        }

        printf("modbus结束\n");
        strcpy(payload, cJSON_Print(root));
        // 发布初始消息
        pubmsg.payload = payload;
        pubmsg.payloadlen = (int)strlen(payload);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;
        MQTTClient_publishMessage(client, TOPICPUB, &pubmsg, &token);
        printf("Waiting for up to %d seconds for publication of %s\n"
               "on topic %s for client with ClientID: %s\n",
               (int)(TIMEOUT / 1000), payload, TOPICPUB, CLIENTID);

        rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
        printf("Message with delivery token %d delivered\n", token);
        cJSON_Delete(root); // 释放初始JSON对象

        // 刷新上报循环
        while (1)
        {
            List_Data *tmp;
            struct list_head *pos;
            int *shm_stm_addr_int = (int *)shm_stm_addr;
            Shm_Stm *shm_stm_addr_struct = (Shm_Stm *)(shm_stm_addr_int + 1);

            // 检查STM数据变化
            list_for_each(pos, &stm_head)
            {
                tmp = list_entry(pos, List_Data, list);
                if (!((tmp->key == shm_stm_addr_struct->key) && (tmp->val.c == shm_stm_addr_struct->val.c)))
                {
                    tmp->val = shm_stm_addr_struct->val;
                    cJSON *root = cJSON_CreateObject(); // 创建临时JSON对象
                    cJSON *item = cJSON_CreateObject();
                    cJSON_AddItemToObject(root, "m0", item);
                    cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(shm_stm_addr_struct->key));
                    // 设置名称
                    switch (shm_stm_addr_struct->key)
                    {
                    case 1: cJSON_AddItemToObject(item, "name", cJSON_CreateString("temperature")); break;
                    case 2: cJSON_AddItemToObject(item, "name", cJSON_CreateString("humidity")); break;
                    case 3: cJSON_AddItemToObject(item, "name", cJSON_CreateString("red_light")); break;
                    case 4: cJSON_AddItemToObject(item, "name", cJSON_CreateString("fire")); break;
                    case 5: cJSON_AddItemToObject(item, "name", cJSON_CreateString("fan")); break;
                    case 6: cJSON_AddItemToObject(item, "name", cJSON_CreateString("window")); break;
                    case 7: cJSON_AddItemToObject(item, "name", cJSON_CreateString("light")); break;
                    case 8: cJSON_AddItemToObject(item, "name", cJSON_CreateString("buzzer")); break;
                    default: break;
                    }
                    cJSON_AddItemToObject(item, "val", cJSON_CreateNumber(shm_stm_addr_struct->val.c));
                    
                    memset(payload, 0, 1024);
                    strcpy(payload, cJSON_Print(root));
                    // 发布变化消息
                    pubmsg.payload = payload;
                    pubmsg.payloadlen = (int)strlen(payload);
                    pubmsg.qos = QOS;
                    pubmsg.retained = 0;
                    MQTTClient_publishMessage(client, TOPICPUB, &pubmsg, &token);
                    printf("Waiting for up to %d seconds for publication of %s\n"
                           "on topic %s for client with ClientID: %s\n",
                           (int)(TIMEOUT / 1000), payload, TOPICPUB, CLIENTID);

                    rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
                    printf("Message with delivery token %d delivered\n", token);
                    cJSON_Delete(root); // 释放临时JSON对象
                }
                shm_stm_addr_struct++;
            }

            // 检查Modbus数据变化
            int *shm_modbus_addr_int = (int *)shm_modbus_addr;
            Shm_Modbus *shm_modbus_addr_struct = (Shm_Modbus *)(shm_modbus_addr_int + 1);
            list_for_each(pos, &modbus_head)
            {
                tmp = list_entry(pos, List_Data, list);
                if (!((tmp->key == shm_modbus_addr_struct->key) & (tmp->val.c == shm_modbus_addr_struct->val.c)))
                {
                    tmp->val = shm_modbus_addr_struct->val;
                    cJSON *root = cJSON_CreateObject(); // 创建临时JSON对象
                    cJSON *item = cJSON_CreateObject();
                    cJSON_AddItemToObject(root, "modbus", item);
                    cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(shm_modbus_addr_struct->key));
                    // 设置名称
                    switch (shm_modbus_addr_struct->key)
                    {
                    case 101: cJSON_AddItemToObject(item, "name", cJSON_CreateString("people")); break;
                    case 102: cJSON_AddItemToObject(item, "name", cJSON_CreateString("door")); break;
                    case 103: cJSON_AddItemToObject(item, "name", cJSON_CreateString("window")); break;
                    case 104: cJSON_AddItemToObject(item, "name", cJSON_CreateString("air_switch")); break;
                    case 105: cJSON_AddItemToObject(item, "name", cJSON_CreateString("air_temp")); break;
                    case 106: cJSON_AddItemToObject(item, "name", cJSON_CreateString("fan")); break;
                    default: break;
                    }
                    cJSON_AddItemToObject(item, "val", cJSON_CreateNumber(shm_modbus_addr_struct->val.c));
                    
                    memset(payload, 0, 1024);
                    strcpy(payload, cJSON_Print(root));
                    // 发布变化消息
                    pubmsg.payload = payload;
                    pubmsg.payloadlen = (int)strlen(payload);
                    pubmsg.qos = QOS;
                    pubmsg.retained = 0;
                    MQTTClient_publishMessage(client, TOPICPUB, &pubmsg, &token);
                    printf("Waiting for up to %d seconds for publication of %s\n"
                           "on topic %s for client with ClientID: %s\n",
                           (int)(TIMEOUT / 1000), payload, TOPICPUB, CLIENTID);

                    rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
                    printf("Message with delivery token %d delivered\n", token);
                    cJSON_Delete(root); // 释放临时JSON对象
                }
                shm_modbus_addr_struct++;
            }
        }
    }
    else if (shm_report_addr_struct->mode == 2) // 周期上报
    {
        printf("周期上报\n");
        char payload[1024];
        while (1)
        {
            memset(payload, 0, 1024);
            cJSON *root = cJSON_CreateObject(); // 创建周期上报JSON对象
            // 处理STM数据
            {
                cJSON *item = cJSON_CreateObject();
                cJSON *arr = cJSON_CreateArray();
                cJSON_AddItemToObject(root, "m0", item);
                cJSON_AddItemToObject(item, "data", arr);

                int *shm_stm_addr_int = (int *)shm_stm_addr;
                Shm_Stm *shm_stm_addr_struct = (Shm_Stm *)(shm_stm_addr_int + 1);
                printf("%d\n", *shm_stm_addr_int);

                for (int i = 0; i < *shm_stm_addr_int; i++)
                {
                    cJSON *data = cJSON_CreateObject();
                    cJSON_AddItemToObject(data, "key", cJSON_CreateNumber(shm_stm_addr_struct->key));
                    // 设置名称
                    switch (shm_stm_addr_struct->key)
                    {
                    case 1: cJSON_AddItemToObject(data, "name", cJSON_CreateString("temperature")); break;
                    case 2: cJSON_AddItemToObject(data, "name", cJSON_CreateString("humidity")); break;
                    case 3: cJSON_AddItemToObject(data, "name", cJSON_CreateString("red_light")); break;
                    case 4: cJSON_AddItemToObject(data, "name", cJSON_CreateString("fire")); break;
                    case 5: cJSON_AddItemToObject(data, "name", cJSON_CreateString("fan")); break;
                    case 6: cJSON_AddItemToObject(data, "name", cJSON_CreateString("window")); break;
                    case 7: cJSON_AddItemToObject(data, "name", cJSON_CreateString("light")); break;
                    case 8: cJSON_AddItemToObject(data, "name", cJSON_CreateString("buzzer")); break;
                    default: break;
                    }
                    cJSON_AddItemToObject(data, "type", cJSON_CreateNumber(shm_stm_addr_struct->type));
                    // 设置值（根据类型）
                    if (shm_stm_addr_struct->type == 1)
                        cJSON_AddItemToObject(data, "val", cJSON_CreateNumber(shm_stm_addr_struct->val.a));
                    else if (shm_stm_addr_struct->type == 2)
                        cJSON_AddItemToObject(data, "val", cJSON_CreateNumber(shm_stm_addr_struct->val.b));
                    else if (shm_stm_addr_struct->type == 3)
                        cJSON_AddItemToObject(data, "val", cJSON_CreateNumber(shm_stm_addr_struct->val.c));
                    
                    cJSON_AddItemToArray(arr, data);
                    shm_stm_addr_struct++;
                }
            }
            // 处理Modbus数据
            {
                cJSON *item = cJSON_CreateObject();
                cJSON *arr = cJSON_CreateArray();
                cJSON_AddItemToObject(root, "modbus", item);
                cJSON_AddItemToObject(item, "data", arr);

                int *shm_modbus_addr_int = (int *)shm_modbus_addr;
                Shm_Modbus *shm_modbus_addr_struct = (Shm_Modbus *)(shm_modbus_addr_int + 1);
                printf("%d\n", *shm_modbus_addr_int);

                for (int i = 0; i < *shm_modbus_addr_int; i++)
                {
                    cJSON *data = cJSON_CreateObject();
                    cJSON_AddItemToObject(data, "key", cJSON_CreateNumber(shm_modbus_addr_struct->key));
                    // 设置名称（修复原代码switch(1)的逻辑错误）
                    switch (shm_modbus_addr_struct->key)
                    {
                    case 101: cJSON_AddItemToObject(data, "name", cJSON_CreateString("people")); break;
                    case 102: cJSON_AddItemToObject(data, "name", cJSON_CreateString("door")); break;
                    case 103: cJSON_AddItemToObject(data, "name", cJSON_CreateString("window")); break;
                    case 104: cJSON_AddItemToObject(data, "name", cJSON_CreateString("air_switch")); break;
                    case 105: cJSON_AddItemToObject(data, "name", cJSON_CreateString("air_temp")); break;
                    case 106: cJSON_AddItemToObject(data, "name", cJSON_CreateString("fan")); break;
                    default: break;
                    }
                    cJSON_AddItemToObject(data, "type", cJSON_CreateNumber(shm_modbus_addr_struct->type));
                    // 设置值（根据类型）
                    if (shm_modbus_addr_struct->type == 1)
                        cJSON_AddItemToObject(data, "val", cJSON_CreateNumber(shm_modbus_addr_struct->val.a));
                    else if (shm_modbus_addr_struct->type == 2)
                        cJSON_AddItemToObject(data, "val", cJSON_CreateNumber(shm_modbus_addr_struct->val.b));
                    else if (shm_modbus_addr_struct->type == 3)
                        cJSON_AddItemToObject(data, "val", cJSON_CreateNumber(shm_modbus_addr_struct->val.c));
                    
                    cJSON_AddItemToArray(arr, data);
                    shm_modbus_addr_struct++;
                }
            }

            strcpy(payload, cJSON_Print(root));
            printf("%s\n", payload);
            // 发布周期消息
            pubmsg.payload = payload;
            pubmsg.payloadlen = (int)strlen(payload);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;
            MQTTClient_publishMessage(client, TOPICPUB, &pubmsg, &token);
            printf("Waiting for up to %d seconds for publication of %s\n"
                   "on topic %s for client with ClientID: %s\n",
                   (int)(TIMEOUT / 1000), payload, TOPICPUB, CLIENTID);

            rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
            printf("Message with delivery token %d delivered\n", token);
            cJSON_Delete(root); // 释放周期上报JSON对象
            sleep(shm_report_addr_struct->period);
        }
    }

    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    return NULL;
}

volatile MQTTClient_deliveryToken deliveredtoken;

// MQTT消息发送确认回调
void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

// MQTT消息接收回调
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    char *payloadptr = (char *)message->payload;
    printf("Message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: %s\n", payloadptr);

    if (strstr(payloadptr, "m0") != NULL)
    {
        cJSON *root = cJSON_Parse(payloadptr);
        if (root)
        {
            cJSON *item = cJSON_GetObjectItem(root, "m0");
            if (item)
            {
                Message msg;
                msg.mtype=1;
                strncpy(msg.mtext, item->valuestring, sizeof(msg.mtext) - 1);
                msg.mtext[sizeof(msg.mtext) - 1] = '\0';
                printf("%s\n", msg.mtext);
                msg_queue_send("stm", &msg, sizeof(msg), 0);
            }
            cJSON_Delete(root); // 释放解析的JSON对象
        }
    }
    else if (strstr(payloadptr, "modbus") != NULL)
    {
        printf("收到modbus\n");
        cJSON *root = cJSON_Parse(payloadptr);
        if (root)
        {
            cJSON *item = cJSON_GetObjectItem(root, "modbus");
            if (item)
            {
                Message msg;
                msg.mtype=1;
                strncpy(msg.mtext, item->valuestring, sizeof(msg.mtext) - 1);
                msg.mtext[sizeof(msg.mtext) - 1] = '\0';
                msg_queue_send("modbus", &msg, sizeof(msg), 0);
            }
            cJSON_Delete(root); // 释放解析的JSON对象
        }
    }
    else if (strstr(payloadptr, "flush_data") != NULL) // 刷新数据
    {
        printf("收到刷新\n");
        cJSON *root = cJSON_Parse(payloadptr);
        if (root)
        {
            cJSON *item = cJSON_GetObjectItem(root, "flush_data");
            if (item && strcmp(item->string, "flush_data") == 0 && item->valueint == 1)
            {
                MQTTClient client;
                MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
                MQTTClient_message pubmsg = MQTTClient_message_initializer;
                MQTTClient_deliveryToken token;
                int rc;

                MQTTClient_create(&client, ADDRESS, CLIENTID_FLUSH,
                                  MQTTCLIENT_PERSISTENCE_NONE, NULL);
                conn_opts.keepAliveInterval = 20;
                conn_opts.cleansession = 1;

                if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
                {
                    printf("Failed to connect, return code %d\n", rc);
                    MQTTClient_destroy(&client);
                    cJSON_Delete(root); // 释放解析的JSON对象
                    goto cleanup;
                }

                char payload[1024];
                cJSON *flush_root = cJSON_CreateObject(); // 创建刷新响应JSON对象
                // 处理STM数据
                {
                    cJSON *item = cJSON_CreateObject();
                    cJSON *arr = cJSON_CreateArray();
                    cJSON_AddItemToObject(flush_root, "m0", item);
                    cJSON_AddItemToObject(item, "data", arr);

                    int *shm_stm_addr_int = (int *)shm_stm_addr;
                    Shm_Stm *shm_stm_addr_struct = (Shm_Stm *)(shm_stm_addr_int + 1);
                    printf("%d\n", *shm_stm_addr_int);

                    for (int i = 0; i < *shm_stm_addr_int; i++)
                    {
                        cJSON *data = cJSON_CreateObject();
                        // 设置名称
                        switch (shm_stm_addr_struct->key)
                        {
                        case 1: cJSON_AddItemToObject(data, "name", cJSON_CreateString("temperature")); break;
                        case 2: cJSON_AddItemToObject(data, "name", cJSON_CreateString("humidity")); break;
                        case 3: cJSON_AddItemToObject(data, "name", cJSON_CreateString("red_light")); break;
                        case 4: cJSON_AddItemToObject(data, "name", cJSON_CreateString("fire")); break;
                        case 5: cJSON_AddItemToObject(data, "name", cJSON_CreateString("fan")); break;
                        case 6: cJSON_AddItemToObject(data, "name", cJSON_CreateString("window")); break;
                        case 7: cJSON_AddItemToObject(data, "name", cJSON_CreateString("light")); break;
                        case 8: cJSON_AddItemToObject(data, "name", cJSON_CreateString("buzzer")); break;
                        default: break;
                        }
                        cJSON_AddItemToObject(data, "key", cJSON_CreateNumber(shm_stm_addr_struct->key));
                        cJSON_AddItemToObject(data, "type", cJSON_CreateNumber(shm_stm_addr_struct->type));
                        // 设置值（根据类型）
                        if (shm_stm_addr_struct->type == 1)
                            cJSON_AddItemToObject(data, "val", cJSON_CreateNumber(shm_stm_addr_struct->val.a));
                        else if (shm_stm_addr_struct->type == 2)
                            cJSON_AddItemToObject(data, "val", cJSON_CreateNumber(shm_stm_addr_struct->val.b));
                        else if (shm_stm_addr_struct->type == 3)
                            cJSON_AddItemToObject(data, "val", cJSON_CreateNumber(shm_stm_addr_struct->val.c));
                        
                        cJSON_AddItemToArray(arr, data);
                        shm_stm_addr_struct++;
                    }
                }
                // 处理Modbus数据
                {
                    cJSON *item = cJSON_CreateObject();
                    cJSON *arr = cJSON_CreateArray();
                    cJSON_AddItemToObject(flush_root, "modbus", item);
                    cJSON_AddItemToObject(item, "data", arr);

                    int *shm_modbus_addr_int = (int *)shm_modbus_addr;
                    Shm_Modbus *shm_modbus_addr_struct = (Shm_Modbus *)(shm_modbus_addr_int + 1);
                    printf("%d\n", *shm_modbus_addr_int);

                    for (int i = 0; i < *shm_modbus_addr_int; i++)
                    {
                        cJSON *data = cJSON_CreateObject();
                        cJSON_AddItemToObject(data, "key", cJSON_CreateNumber(shm_modbus_addr_struct->key));
                        // 设置名称（修复原代码switch(1)的逻辑错误）
                        switch (shm_modbus_addr_struct->key)
                        {
                        case 101: cJSON_AddItemToObject(data, "name", cJSON_CreateString("people")); break;
                        case 102: cJSON_AddItemToObject(data, "name", cJSON_CreateString("door")); break;
                        case 103: cJSON_AddItemToObject(data, "name", cJSON_CreateString("window")); break;
                        case 104: cJSON_AddItemToObject(data, "name", cJSON_CreateString("air_switch")); break;
                        case 105: cJSON_AddItemToObject(data, "name", cJSON_CreateString("air_temp")); break;
                        case 106: cJSON_AddItemToObject(data, "name", cJSON_CreateString("fan")); break;
                        default: break;
                        }
                        cJSON_AddItemToObject(data, "type", cJSON_CreateNumber(shm_modbus_addr_struct->type));
                        // 设置值（根据类型）
                        if (shm_modbus_addr_struct->type == 1)
                            cJSON_AddItemToObject(data, "val", cJSON_CreateNumber(shm_modbus_addr_struct->val.a));
                        else if (shm_modbus_addr_struct->type == 2)
                            cJSON_AddItemToObject(data, "val", cJSON_CreateNumber(shm_modbus_addr_struct->val.b));
                        else if (shm_modbus_addr_struct->type == 3)
                            cJSON_AddItemToObject(data, "val", cJSON_CreateNumber(shm_modbus_addr_struct->val.c));
                        
                        cJSON_AddItemToArray(arr, data);
                        shm_modbus_addr_struct++;
                    }
                }

                strcpy(payload, cJSON_Print(flush_root));
                // 发布刷新响应
                pubmsg.payload = payload;
                pubmsg.payloadlen = (int)strlen(payload);
                pubmsg.qos = QOS;
                pubmsg.retained = 0;
                MQTTClient_publishMessage(client, TOPICPUB, &pubmsg, &token);
                printf("Waiting for up to %d seconds for publication of %s\n"
                       "on topic %s for client with ClientID: %s\n",
                       (int)(TIMEOUT / 1000), payload, TOPICPUB, CLIENTID_FLUSH);

                rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
                printf("Message with delivery token %d delivered\n", token);
                cJSON_Delete(flush_root); // 释放刷新响应JSON对象
                MQTTClient_disconnect(client, 10000);
                MQTTClient_destroy(&client);
            }
            cJSON_Delete(root); // 释放解析的JSON对象
        }
    }
    else if (strstr(payloadptr, "report") != NULL) // 上报方式配置
    {
        printf("收到上报\n");
        cJSON *root = cJSON_Parse(payloadptr);
        if (root)
        {
            cJSON *item = cJSON_GetObjectItem(root, "report");
            if (item)
            {
                cJSON *mode = cJSON_GetObjectItem(item, "mode");
                cJSON *val = cJSON_GetObjectItem(item, "period");
                FILE *json = fopen("./json.json", "w");
                if (json != NULL)
                {
                    fseek(json, 47, SEEK_SET);
                    int mode_valueint = mode ? mode->valueint : 0;
                    fwrite(&mode_valueint, sizeof(int), 1, json);
                    if (mode && mode->valueint == 2)
                    {
                        Shm_Report *shm_report_addr_struct = (Shm_Report *)shm_report_addr;
                        shm_report_addr_struct->period = val ? val->valueint : 0;
                    }
                    fclose(json);
                }
                else
                {
                    perror("open json file err");
                }
            }
            cJSON_Delete(root); // 释放解析的JSON对象
        }
    }

cleanup:
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

// MQTT连接丢失回调
void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

// 控制线程（订阅MQTT消息）
void *pthread_control(void *arg)
{
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;

    MQTTClient_create(&client, ADDRESS, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
           "Press Q<Enter> to quit\n\n",
           TOPICSUB, CLIENTID, QOS);
    MQTTClient_subscribe(client, TOPICSUB, QOS);

    // 阻塞等待消息（实际应用中可通过信号退出）
    pause();

    MQTTClient_unsubscribe(client, TOPICSUB);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    return NULL;
}